package com.mc.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.List;

/***
 * @ClassName UserConsumer
 * @Description todo
 * @Author CC
 * @Date: 2019/11/1215:12
 * @Version 1.0
 */
@Service
public class UserConsumer {

    @Value("${rocketmq.consumerGroup}")
    private String consumerGroup;

    /**
     * NameServer 地址
     */
    @Value("${rocketmq.namesrvaddr}")
    private String namesrvAddr;

    @PostConstruct
    public void consumer() throws MQClientException {
        System.err.println("init defaultMQPushConsumer");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.subscribe("user-topic", "user-tag");
        consumer.registerMessageListener((List<MessageExt> list, ConsumeConcurrentlyContext context) -> {
            for (MessageExt messageExt : list) {
                System.err.println("消费消息: " + new String(messageExt.getBody()));//输出消息内容
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
        });
        consumer.start();
    }

}
